import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.common.operators.Order;
public class GroupReduceonsortedgroups {


    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String rootPath = new StringBuilder(System.getProperty("user.dir")).toString();
        DataSet<Tuple2<Integer, String>> input = env.readCsvFile("file:///"+rootPath+"/"+"group.csv").types(Integer.class, String.class);


//        input.print();
//        System.out.println("-----------------------input------------------------");


        DataSet<Tuple2<Integer,String>> output2 = input.groupBy(0).getInputDataSet();                         // group DataSet by first field

        output2.print();
        System.out.println("-----------------------output2------------------------");

        DataSet<Tuple2<Integer,String>> output1 = input.groupBy(0).sortGroup(1, Order.ASCENDING).getInputDataSet();      // sort groups on second tuple field
        output1.print();
        System.out.println("------------------------output1-----------------------");
        DataSet<Tuple2<Integer,String>> output = output2.reduceGroup(new DistinctReduce2());  //对每组配合进行去重



        output.print();


    }
}